package com.relytosoft.mizar.analyze.face.services.service.impl;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.relytosoft.dubhe.common.core.base.page.PageQuery;
import com.relytosoft.dubhe.common.core.exception.ServiceException;
import com.relytosoft.dubhe.common.core.object.ResponseResult;
import com.relytosoft.dubhe.common.core.utils.BeanCopyUtils;
import com.relytosoft.dubhe.common.elasticsearch.ElasticsearchUtil;
import com.relytosoft.dubhe.common.elasticsearch.page.PageResult;
import com.relytosoft.dubhe.common.kafka.producer.KafkaOpMsg;
import com.relytosoft.dubhe.common.kafka.producer.KafkaProducer;
import com.relytosoft.dubhe.common.redis.utils.RedisUtil;
import com.relytosoft.mizar.analyze.base.constant.AyCacheConstants;
import com.relytosoft.mizar.analyze.base.constant.MediaMqConstants;
import com.relytosoft.mizar.analyze.base.entity.pojo.dto.AyGroupBaseDto;
import com.relytosoft.mizar.analyze.base.entity.pojo.dto.AyPlanTemplateDto;
import com.relytosoft.mizar.analyze.base.entity.pojo.dto.DispositionReqDto;
import com.relytosoft.mizar.analyze.base.enums.AyFaceEnum;
import com.relytosoft.mizar.analyze.base.enums.IndexNameTypeEnum;
import com.relytosoft.mizar.analyze.common.feign.SystemFeignService;
import com.relytosoft.mizar.analyze.common.utils.Base64Utils;
import com.relytosoft.mizar.analyze.common.utils.IndexUtil;
import com.relytosoft.mizar.analyze.face.api.entity.AyFaceDisposition;
import com.relytosoft.mizar.analyze.face.api.pojo.dto.AyFaceDispositionDto;
import com.relytosoft.mizar.analyze.face.api.pojo.query.AyFaceSearchQuery;
import com.relytosoft.mizar.analyze.face.api.pojo.vo.AyFaceDispositionVo;
import com.relytosoft.mizar.analyze.face.services.mapper.AyFaceDispositionMapper;
import com.relytosoft.mizar.analyze.face.services.service.AyFaceDispositionService;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.*;
import java.util.stream.Collectors;


/**
 * @author lj
 * @date 2022/5/16 12:00
 */
@Service
public class AyFaceDispositionServiceImpl extends ServiceImpl<AyFaceDispositionMapper, AyFaceDisposition> implements AyFaceDispositionService {

    @Resource
    private KafkaProducer kafkaProducer;

    @Resource
    private ElasticsearchUtil elasticsearchUtil;

    @Resource
    private SystemFeignService systemFeignService;

    @Override
    public IPage<AyFaceDisposition> selectAyFaceDispositionPageList(PageQuery pageQuery, AyFaceDisposition ayFaceDisposition) {
        if (StrUtil.isBlank(pageQuery.getOrderByColumn())) {
            pageQuery.setOrderByColumn("id");
            pageQuery.setIsAsc("desc");
        }
        return baseMapper.selectPageDispositionList(pageQuery.build(), ayFaceDisposition);
    }

    @Override
    public List<AyFaceDisposition> addAyFaceDisposition(List<AyFaceDisposition> ayFaceDispositionList) {
        boolean saveBatch = this.saveBatch(ayFaceDispositionList);
        if (saveBatch) {
            return ayFaceDispositionList;
        }
        return new ArrayList<>();
    }

    @Override
    public void sendAddInfoToKafka(List<AyFaceDisposition> ayFaceDispositionList) {
        ayFaceDispositionList.forEach(this::sendCreateToKafka);
    }

    @Override
    public List<AyFaceDisposition> getAyFaceDispositionInfo(AyFaceDispositionDto ayFaceDispositionDto) {
        Integer planTemplateId = ayFaceDispositionDto.getPlanTemplateId();
        //获取计划模板信息
        AyPlanTemplateDto planTemplateDto = null;
        if (planTemplateId != null) {
            planTemplateDto = getPlanTemplate(planTemplateId);
        }
        //获取需要添加布控任务得设备ID
        String[] dispositionDeviceIds = ayFaceDispositionDto.getDispositionRangeDevice().split(",");
        List<AyFaceDisposition> list = new ArrayList<>();
        AyPlanTemplateDto finalPlanTemplateDto = planTemplateDto;
        Arrays.stream(dispositionDeviceIds).forEach(deviceId -> {
            AyFaceDisposition ayFaceDisposition = BeanCopyUtils.copy(ayFaceDispositionDto, AyFaceDisposition.class);
            //设置必要参数
            ayFaceDisposition.setDispositionRangeDevice(deviceId);
            ayFaceDisposition.setStatus(0);
            ayFaceDisposition.setDefenseStatus(0);
            ayFaceDisposition.setDispositionId(UUID.randomUUID().toString());
            if (finalPlanTemplateDto != null) {
                ayFaceDisposition.setPlanType(finalPlanTemplateDto.getPlanType());
                ayFaceDisposition.setPlanInfoNum(finalPlanTemplateDto.getPlanInfoNum());
                ayFaceDisposition.setPlanInfoList(finalPlanTemplateDto.getPlanInfoList());
            }
            list.add(ayFaceDisposition);
        });
        return list;
    }


    @Override
    public AyFaceDispositionVo selectAyFaceDispositionById(Integer faceDispositionId) {
        AyFaceDisposition ayFaceDisposition = this.getBaseMapper().selectById(faceDispositionId);
        return BeanCopyUtils.copy(ayFaceDisposition, AyFaceDispositionVo.class);
    }


    @Override
    public boolean updateAyFaceDisposition(AyFaceDispositionDto ayFaceDispositionDto) {
        //查询原来数据的状态信息
        AyFaceDisposition disposition = this.getById(ayFaceDispositionDto.getId());
        //当布控状态为未开始或已结束或异常
        if (disposition.getStatus() != 1) {
            AyFaceDisposition ayFaceDisposition = BeanCopyUtils.copy(ayFaceDispositionDto, AyFaceDisposition.class);
            //如果模板为空直接添加
            if (ayFaceDispositionDto.getPlanTemplateId() != null) {
                getUpdateAyFaceDispositionInfo(ayFaceDispositionDto, ayFaceDisposition);
            }
            //更新本库数据信息
            boolean update = this.updateById(ayFaceDisposition);
            //2. 删除比特大陆布控任务
            sendToDeleteToKafka(ayFaceDisposition.getDispositionId());
            sendCreateToKafka(ayFaceDisposition);
            return update;
        }
        return false;
    }


    /**
     * 设置模板信息
     *
     * @param ayFaceDispositionDto 修改模板信息
     */
    private void getUpdateAyFaceDispositionInfo(AyFaceDispositionDto ayFaceDispositionDto, AyFaceDisposition ayFaceDisposition) {
        //获取计划模板信息
        AyPlanTemplateDto planTemplate = getPlanTemplate(ayFaceDispositionDto.getPlanTemplateId());
        //设置必要参数
        ayFaceDisposition.setDispositionRangeDevice(ayFaceDispositionDto.getDispositionRangeDevice());
        ayFaceDisposition.setPlanType(planTemplate.getPlanType());
        ayFaceDisposition.setPlanInfoNum(planTemplate.getPlanInfoNum());
        ayFaceDisposition.setPlanInfoList(planTemplate.getPlanInfoList());
    }

    /**
     * 远程查询模板信息
     *
     * @param planTemplateId 计划模板ID
     * @return AyPlanTemplateDto
     */
    private AyPlanTemplateDto getPlanTemplate(Integer planTemplateId) {
        ResponseResult result = systemFeignService.id(planTemplateId);
        String jsonString = JSONObject.toJSONString(result.getData());
        return JSON.parseObject(jsonString, AyPlanTemplateDto.class);
    }


    @Override
    public Map<Integer, String> removeAyFaceDisposition(List<Integer> faceDispositionIds) {
        //1. 判断是否为已停止和撤防状态
        List<AyFaceDisposition> list = this.listByIds(faceDispositionIds);
        Map<Integer, String> map = new HashMap<>(66);
        Set<Integer> set = list.stream().filter(item -> {
            if (item.getStatus() != 1 && item.getDefenseStatus() == 0) {
                return true;
            } else {
                map.put(item.getId(), item.getName() + "-处于未停止未撤防状态不能删除");
                return false;
            }
        }).map(AyFaceDisposition::getId).collect(Collectors.toSet());
        //删除
        this.removeBatchByIds(set);
        //获取ID和比特大陆布控ID映射关系
        Map<Integer, String> idAndDispositionId = list.stream().collect(Collectors.toMap(AyFaceDisposition::getId,
                AyFaceDisposition::getDispositionId));
        //发送删除消息到kafka
        set.forEach(id -> sendToDeleteToKafka(idAndDispositionId.get(id)));
        //处于布防或布控状态不能删除
        return map;
    }


    /**
     * 相当于删除比特大陆原来的任务 创建一个新任务
     *
     * @param id 需要开始的的任务id
     * @return 是否成功
     */
    @Override
    public String startTask(Integer id) {
        //1. 查询布控任务信息
        AyFaceDisposition ayFaceDisposition = this.getById(id);
        //2.判断布控任务状态
        if (ayFaceDisposition.getStatus() == 0 || ayFaceDisposition.getStatus() == 2) {
            //删除任务
            sendToDeleteToKafka(ayFaceDisposition.getDispositionId());
            //在创建任务
            sendCreateToKafka(ayFaceDisposition);
            return "启动成功";
        }
        if (ayFaceDisposition.getStatus() == 1) {
            return "任务正常运行中无需启动";
        }
        if (ayFaceDisposition.getStatus() == 3) {
            return "任务运行异常，请联系管理员";
        }
        return "启动任务失败";
    }

    /**
     * 相当于删除比特大陆的任务
     *
     * @param id 需要停止的任务id
     */
    @Override
    public boolean stopTask(Integer id) {
        //1. 查询布控任务信息
        AyFaceDisposition ayFaceDisposition = this.getById(id);

        if (ayFaceDisposition == null) {
            throw new ServiceException("任务不存在");
        }
        //2.判断布控任务状态
        if (ayFaceDisposition.getStatus() == 1) {
            //发送删除的消息
            sendToDeleteToKafka(ayFaceDisposition.getDispositionId());
            ayFaceDisposition.setDefenseStatus(0);
            ayFaceDisposition.setStatus(2);
            ayFaceDisposition.setEndTime(LocalDateTime.now());
            return this.updateById(ayFaceDisposition);
        } else {
            return false;
        }
    }

    @Override
    public boolean armedAndDisarmed(List<Integer> list) {
        List<AyFaceDisposition> ayFaceDispositions = this.listByIds(list);
        ayFaceDispositions.forEach(item -> {
            item.setDefenseStatus(item.getDefenseStatus() == 0 ? 1 : 0);
            //更新redis布防撤防信息
            RedisUtil.set(AyCacheConstants.CONTROL_FACE_CONDITION + item.getDispositionId(), JSON.toJSONString(item));
        });
        return this.updateBatchById(ayFaceDispositions);
    }

    @Override
    public JSONObject faceSearch(AyFaceSearchQuery ayFaceSearchQuery, PageQuery pageQuery) {
        String index = IndexUtil.toMonthIndex(IndexNameTypeEnum.AY_FACE.getMessage());
        BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();
        if (ayFaceSearchQuery.getDeviceId() != null) {
            boolQueryBuilder.must(QueryBuilders.termsQuery("deviceId", ayFaceSearchQuery.getDeviceId().split(",")));
        }
        //根据性别查询
        if (ayFaceSearchQuery.getSex() != null) {
            boolQueryBuilder.must(QueryBuilders.termsQuery("genderCode", ayFaceSearchQuery.getSex().split(",")));
        }
        //墨镜
        if (ayFaceSearchQuery.getGlasses() != null) {
            boolQueryBuilder.must(QueryBuilders.termsQuery("glassExtend", ayFaceSearchQuery.getGlasses().split(",")));
        }
        //微笑
        if (ayFaceSearchQuery.getSmile() != null) {
            boolQueryBuilder.must(QueryBuilders.termsQuery("smile", ayFaceSearchQuery.getSmile().split(",")));
        }
        //眼镜
        if (ayFaceSearchQuery.getSunglasses() != null) {
            boolQueryBuilder.must(QueryBuilders.termsQuery("sunGlassExtend", ayFaceSearchQuery.getSunglasses().split(",")));
        }
        //年龄
        if (ayFaceSearchQuery.getAge() != null) {
            boolQueryBuilder.must(QueryBuilders.termsQuery("age", ayFaceSearchQuery.getAge().split(",")));
        }
        if (ayFaceSearchQuery.getStartTime() != null) {
            boolQueryBuilder.must(QueryBuilders.rangeQuery("markTime").gte(ayFaceSearchQuery.getStartTime().getTime()));
        }
        if (ayFaceSearchQuery.getEndTime() != null) {
            boolQueryBuilder.must(QueryBuilders.rangeQuery("markTime").lte(ayFaceSearchQuery.getEndTime().getTime()));
        }
        try {
            PageResult pageResult = elasticsearchUtil.searchPage(pageQuery, new String[]{index}, boolQueryBuilder);
            return JSON.parseObject(JSON.toJSONString(pageResult));
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public boolean addTempFace(AyFaceDispositionDto ayFaceDispositionDto) {
        //查询
        AyFaceDisposition ayFaceDisposition = this.getById(ayFaceDispositionDto.getId());
        //设置临时人脸输出
        if (ayFaceDisposition.getAlarmType() == 0) {
            ayFaceDisposition.setTempFaceGroup(ayFaceDispositionDto.getTempFaceGroup());
            ayFaceDisposition.setTempBeginTime(getTime(ayFaceDispositionDto.getTempBeginTime()));
            ayFaceDisposition.setTempEndTime(getTime(ayFaceDispositionDto.getTempEndTime()));
            //更新数据库
            boolean update = this.updateById(ayFaceDisposition);
            if (update) {
                //发送删除
                sendToDeleteToKafka(ayFaceDisposition.getDispositionRangeDevice());
                //新建
                sendCreateToKafka(ayFaceDisposition);
                return update;
            }
        }
        return false;
    }

    LocalDateTime getTime(String time) {
        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        Date date = null;
        try {
            date = dateFormat.parse(time);
        } catch (ParseException e) {
            e.printStackTrace();
        }
        assert date != null;
        Instant instant = date.toInstant();
        ZoneId zoneId = ZoneId.systemDefault();
        return instant.atZone(zoneId).toLocalDateTime();
    }

    public void sendCreateToKafka(AyFaceDisposition ayFaceDisposition) {
        DispositionReqDto dispositionReqDto = createDispositionReqDto(ayFaceDisposition);
        //发送消息
        KafkaOpMsg kafkaOpMsg = new KafkaOpMsg();
        kafkaOpMsg.setOpType("insert");
        kafkaOpMsg.setData(JSONObject.toJSONString(dispositionReqDto));
        kafkaProducer.send(MediaMqConstants.FACE_DISPOSITION_TOPIC, JSONObject.toJSONString(kafkaOpMsg), true);
    }

    public DispositionReqDto createDispositionReqDto(AyFaceDisposition ayFaceDisposition) {
        //2. 设置添加任务必要参数
        DispositionReqDto dispositionReqDto = new DispositionReqDto();
        dispositionReqDto.setDeviceId(ayFaceDisposition.getDispositionRangeDevice());
        dispositionReqDto.setId(ayFaceDisposition.getId());
        // 人脸库布控
        if (ayFaceDisposition.getDispositionType().equals(AyFaceEnum.TIAOJIAN.getCode())) {
            dispositionReqDto.setType(AyFaceEnum.TIAOJIAN.getCode());
            //查询正常布控三方人脸库ID
            ArrayList<String> groupIds = getTripartiteLibraryId(ayFaceDisposition.getFaceGroup());
            if (groupIds.isEmpty()) {
                throw new ServiceException("未获取到三方设备信息");
            }
            //白名单布控
            if (ayFaceDisposition.getAlarmType() == 0) {
                int integer = Integer.parseInt(ayFaceDisposition.getThreshold());
                dispositionReqDto.setThreshold((double) (100 - integer));
                //判断是否临时人脸布控
                if (ayFaceDisposition.getTempFaceGroup() != null) {
                    ArrayList<String> tempGroupIds = getTripartiteLibraryId(ayFaceDisposition.getTempFaceGroup());
                    groupIds.addAll(tempGroupIds);
                }
            }
            //黑名单布控
            if (ayFaceDisposition.getAlarmType() == 1) {
                dispositionReqDto.setThreshold(Double.valueOf(ayFaceDisposition.getThreshold()));
            }
            dispositionReqDto.setGroupIds(groupIds);
        }
        //图片布控
        if (ayFaceDisposition.getDispositionType().equals(AyFaceEnum.TUPIAN.getCode())) {
            dispositionReqDto.setThreshold(Double.valueOf(ayFaceDisposition.getThreshold()));
            dispositionReqDto.setType(AyFaceEnum.TUPIAN.getCode());
            dispositionReqDto.setFeatureData(Base64Utils.picToBase64(ayFaceDisposition.getImageUrl()));
        }
        return dispositionReqDto;
    }

    public void sendToDeleteToKafka(String dispositionId) {
        //发送消息至kafka启动
        KafkaOpMsg kafkaOpMsg = new KafkaOpMsg();
        kafkaOpMsg.setOpType("delete");
        kafkaOpMsg.setData(dispositionId);
        kafkaProducer.send(MediaMqConstants.FACE_DISPOSITION_TOPIC, JSONObject.toJSONString(kafkaOpMsg), true);
        RedisUtil.del(AyCacheConstants.CONTROL_FACE_CONDITION + dispositionId);
    }

    /**
     * 远程获取库信息
     *
     * @param faceGroup 库
     * @return 需要布控的三方库ID
     */
    private ArrayList<String> getTripartiteLibraryId(String faceGroup) {
        //1. 逗号切割
        String[] split = faceGroup.split(",");
        //2. 转为integer
        List<Integer> list = Arrays.stream(split).map(Integer::valueOf).collect(Collectors.toList());
        //3. 查询自己数据库对应的三方库ID
        ResponseResult result = systemFeignService.selectGroupByIds(JSONObject.toJSONString(list));
        String jsonString = JSONObject.toJSONString(result.getData());
        List<AyGroupBaseDto> ayGroupBaseDtoList = JSONObject.parseObject(jsonString, new TypeReference<List<AyGroupBaseDto>>() {
        });
        ArrayList<String> arrayList =
                ayGroupBaseDtoList.stream().map(AyGroupBaseDto::getPlateformGroupId).collect(Collectors.toCollection(ArrayList::new));
//        Integer type = ayGroupBaseDtoList.get(0).getGroupType();
//        JSONObject jsonObject = new JSONObject();
//        jsonObject.put("type", type);
//        jsonObject.put("arrayList", JSONObject.toJSONString(arrayList));
        return arrayList;
    }
}